Description

Two new APIs let application code observe which source records a live query is consuming, so callers can subscribe/unsubscribe at an underlying sync layer in lockstep with query usage.

What's added

type TrackedSourceRecord = {
  collectionId: string
  key: string | number
}

type TrackedSourceRecordsChange = {
  added: Array<TrackedSourceRecord>
  removed: Array<TrackedSourceRecord>
}

type SubscribeTrackedSourceRecordsOptions = {
  includeInitialState?: boolean
}

Methods on every Collection:

collection.getTrackedSourceRecords(): Array<TrackedSourceRecord>
collection.subscribeTrackedSourceRecords(
  callback: (change: TrackedSourceRecordsChange) => void,
  options?: SubscribeTrackedSourceRecordsOptions,
): () => void

Same method, polymorphic dispatch:

• On a live-query collection → "records this query reads from its sources." Refcounted across the query's aliases (self-joins are deduped). Visible only while the query has active subscribers.
• On any other collection (base, electric, query-db, localStorage, sqlite-persisted, etc.) → "records of mine consumed by some live query." Refcounted across all live queries that depend on this collection. If two queries use key K, the base view sees K once; only when the last query stops using K does it emit removed.

Why

Useful for custom sync layers that subscribe/unsubscribe at the record level. They need to subscribe when a record becomes actively used by queries and unsubscribe when the last query stops using it. Base-collection residency is too broad — it covers everything sync'd, whether queries want it or not. These APIs provide the narrower "actively used by live queries" signal.

How an app developer uses this

On a single live query (per-query scope):

function ActiveUsersList() {
  const activeUsers = useMemo(
    () =>
      createLiveQueryCollection((q) =>
        q.from({ user: usersCollection }).where(({ user }) => eq(user.active, true)),
      ),
    [],
  )

  // The hook drives sync (it calls subscribeChanges internally), so the
  // tracked-source listener will start receiving deltas as the live query
  // mounts and reads from its sources.
  useEffect(() => {
    return activeUsers.subscribeTrackedSourceRecords(({ added, removed }) => {
      for (const r of added) mySyncLayer.subscribe(r.collectionId, r.key)
      for (const r of removed) mySyncLayer.unsubscribe(r.collectionId, r.key)
    })
  }, [activeUsers])

  const { data } = useLiveQuery(activeUsers)
  return <ul>{data?.map((u) => <li key={u.id}>{u.name}</li>)}</ul>
}

On a base collection (union scope across all live queries):

const unsubscribe = usersCollection.subscribeTrackedSourceRecords(
  ({ added, removed }) => {
    // Fires only when the first query starts tracking a key (0→1),
    // or the last one stops (1→0).
    for (const r of added) mySyncLayer.subscribe(r.collectionId, r.key)
    for (const r of removed) mySyncLayer.unsubscribe(r.collectionId, r.key)
  },
  { includeInitialState: true },
)

The same pattern works for any collection that uses CollectionImpl under the hood — electric, query-db, localStorage, sqlite-persisted, etc. The feature is sync-backend-agnostic because tracking happens in the live-query pipeline (the IVM), not in the sync config.

Behaviour notes:

• includeInitialState: true replays the current snapshot as added on subscribe (if any records are currently tracked and exposed).
• 0 → 1+ subscribers (a live query gains its first consumer): the per-query view emits added for every key the query is reading; the base view emits added for any key reaching refcount 1.
• 1+ → 0 subscribers (last consumer leaves): the per-query view emits removed for every key it was reading; the base view emits removed for any key whose refcount drops to 0.
• Stable-membership updates (an ordered query updates a row that stays in range) emit nothing.
• must-refetch / truncate-refetch cycles that end with the same keys still tracked emit nothing.

How this works internally

Three concentric scopes, each with a real owner.

Scope 1 — alias-level: CollectionSubscriber.sentToD2Keys

For every alias in a live query (e.g. { user: usersCollection }), one CollectionSubscriber is created. It already maintained sentToD2Keys: Set<string | number> to deduplicate inserts into the D2 dataflow graph — the authoritative per-alias record of "keys this alias is feeding the query."

This PR turns that set's mutations into events. On each sendChangesToPipeline, after filterDuplicateInserts has mutated the set, the subscriber derives a net delta from the filtered change stream itself — every surviving insert is a 0→1 transition for its key, every surviving delete is a 1→0 transition. Insert/delete pairs for the same key within one batch (emitted by splitUpdates for stable-membership ordered updates) net to zero. Cost is O(|batch|) — no dependency on the size of sentToD2Keys. The result is emitted via a single public callback property assigned 1:1 by the builder.

Scope 2 — per-live-query: LiveQueryTrackedSourceRecordsAggregator

One aggregator per sync session. Owns a nested refcount map:

private readonly entries = new Map<string, Map<string | number, { refCount: number }>>()
private exposed = false

Outer key is collectionId, inner key is the source record's primitive key — no composite serialization. The refcount is over aliases: a self-join like { employee: emp, manager: emp } has two subscribers emitting independently, and the aggregator dedupes. apply(collectionId, added, removed):

1. Refcount the adds/removes against the per-collection inner map. 0→1 transitions go into netAdded, 1→0 into netRemoved. Empty inner buckets are pruned.
2. If exposed: propagate net transitions to sourceCollections[collectionId]._trackedSourceRecords.apply(...) (scope 3), and fan out directly to the live-query-local listener set (held by reference from the builder).

setExposed(boolean) is the lifecycle gate. false → true replays the snapshot as added. true → false replays as removed. The builder wires this to subscribers:change:

config.collection.on('subscribers:change', (event) => {
  trackedSourceRecordsAggregator.setExposed(event.subscriberCount > 0)
})

Scope 3 — per-base-collection: TrackedSourceRecordsManager

Each CollectionImpl owns one _trackedSourceRecords: TrackedSourceRecordsManager<TKey> field. Refcount map is Map<TKey, { key, refCount }> over live queries. On 0↔1 transitions, listeners registered via subscribe see added / removed. Record-object allocation is skipped when the listener set is empty.

Polymorphic dispatch on Collection.subscribeTrackedSourceRecords

CollectionImpl has an optional _liveQueryTrackedSourceView field. It's undefined on every collection except live queries; bridgeToCreateCollection sets it at construction time to a stable adapter on CollectionConfigBuilder (which routes through whatever aggregator the current sync session has). The public methods on CollectionImpl:

public getTrackedSourceRecords(): Array<TrackedSourceRecord> {
  return this._liveQueryTrackedSourceView?.snapshot()
    ?? this._trackedSourceRecords.get()
}

public subscribeTrackedSourceRecords(callback, options?): () => void {
  if (this._liveQueryTrackedSourceView) {
    return this._liveQueryTrackedSourceView.subscribe(callback, options)
  }
  return this._trackedSourceRecords.subscribe(callback, options)
}

So callers don't need to branch on collection type. Same method name, scope determined by what kind of collection you're holding.

The live-query listener set lives on CollectionConfigBuilder (long-lived). The aggregator holds it by reference. External subscribers attached before the first sync session, or across session start/stop cycles, do not need to re-attach.

End-to-end flow:

CollectionSubscriber
  sentToD2Keys mutates (filterDuplicateInserts)
    count inserts/deletes in filtered batch → net delta
    → onTrackedKeysChange({ added, removed })            [direct callback, 1:1]
      ↓
LiveQueryTrackedSourceRecordsAggregator
  apply(collectionId, added, removed)
    refcount per (collectionId → key) in nested map
    if exposed:
      → sourceCollections[id]._trackedSourceRecords.apply(netAdded, netRemoved)
      if liveQuery listeners.size > 0:
        for listener of listeners: listener({ added: records, removed: records })
                              ↑
                              liveQuery.subscribeTrackedSourceRecords
                              (held by reference; survives sync sessions)
        ↓
TrackedSourceRecordsManager (on each source collection)
  apply(netAdded, netRemoved)
    refcount per key
    if 0↔1 transitions AND listeners.size > 0:
      for listener of listeners: listener({ added: records, removed: records })
                              ↑
                              baseCollection.subscribeTrackedSourceRecords

Performance characteristics

All hot paths are bounded by batch size, with no dependency on the total tracked-set size:

• Scope 1 delta: O(|batch|) via insert/delete count netting on the filtered stream. No copy of sentToD2Keys.
• Scope 2 refcount: O(|batch|) using a nested Map<collectionId, Map<key, Entry>> — primitive keys, no serialization.
• Scope 3 refcount: O(|batch|) using Map<TKey, Entry> — primitive keys, no serialization.
• Record-object allocation at scopes 2 and 3 is gated on a non-empty listener set. Pure-internal batches (refcount updates with no app-code subscribers) skip the object materialization entirely.

Why not just expose CollectionSubscription.subscribedKeys?

Two reasons:

1. sentToD2Keys is per-alias. A live query with multiple aliases on the same collection (self-joins) would require consumers to dedupe. Scope 2 exists precisely to handle that.
2. subscribeChanges buffers truncate/refetch to hide transient empty states. subscribeTrackedSourceRecords needs to emit net membership. Deriving one from the other would mix concerns.

Three scopes with clear ownership match the three semantic views application code may want.

What's not changed

The feature is purely additive on CollectionImpl. Collection.utils and createCollection are byte-identical to upstream — no in-place mutation of user-supplied utils, no merge-precedence changes. The live-query path's utils merge stays at upstream's { ...options.utils, ...config.utils } (config wins). Tracked-source helpers are not on utils at all; they're direct methods on CollectionImpl.

How was this change tested?

• Automated test (unit, integration, etc.)
• Manual test (provide reproducible testing steps below)

Regression coverage in packages/db/tests/query/live-query-collection.test.ts:

• Base-view delta flow with includeInitialState, plus snapshot-as-removed on last-subscriber unsubscribe.
• Base-view ref-counting across overlapping live queries (key not removed until the last query stops using it).
• Stable-membership updates emit nothing (ordered query, in-range update).
• Truncate-refetch cycles that keep the same keys tracked emit nothing.
• Live-query-local view: delta flow + snapshot-as-removed on unsubscribe.
• Live-query-local view: includeInitialState replay.
• Polymorphic-scope test: two live queries with disjoint where-clauses on the same base, asserting the base view sees the union and each per-query view sees only its own scope.

Plus a regression test in packages/db/tests/utility-exposure.test.ts pinning that user-supplied utils keys override built-ins on collision (the upstream merge-precedence contract).

The feature lives entirely on CollectionImpl and the live-query pipeline (CollectionSubscriber → aggregator → manager) — none of the sync backends are touched, so the in-memory tests exercise the same code paths that electric, query-db, localStorage, etc. would hit.

Screenshots